package com.lee.producer;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;

import java.util.List;
import java.util.Objects;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.stream.IntStream;

/**
 * @Description TODO
 * @Author winston
 * @DateTime 2021/11/26
 */
public class MyProducer {
	public static void main(String[] args) {
		// 初始化配置
		Properties properties = initKafka();
//		KafkaProducer

		// 创建生产者对象
		KafkaProducer<String, String> kafkaProducer = new KafkaProducer<>(properties);

		String topic = "testTopicNew2";

		IntStream.rangeClosed(1, 10).forEach(i->{
			String message = "Hello world_" + i;
			normalSend(kafkaProducer, topic, message);
//			producerSendWithCallback(kafkaProducer, topic, message);
//			producerSendWithCallbackWithPartition(kafkaProducer, topic, message);

			System.out.println(message);
		});
		kafkaProducer.flush();
		kafkaProducer.close();// 会调用拦截器中的close方法

	}

	/**
	 * 初始化卡夫卡
	 *
	 * @return {@link Properties}
	 */
	public static Properties initKafka() {
		Properties properties = new Properties();
		// dev环境
//		properties.put("bootstrap.servers", "10.18.0.73:31090,10.18.0.73:31091,10.18.0.73:31092");
		// 本地
		properties.put("bootstrap.servers", "127.0.0.1:9092");
		properties.put("acks", "all");
		properties.put("retries", 3);
		properties.put("batch.size", 16384);
//		等待时间, 默认值60000MS, 一分钟, 坑爹哦.
//		properties.put("max.block.ms", 100);
		properties.put("buffer.memory", 33554432);
		properties.put("linger.ms", 1);
		properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
		// 自定义分区实现类.
		properties.put("partitioner.class", "com.lee.config.MyPartitioner");

		List<String> incerceptors = List.of("com.lee.config.interceptor.TmeInterceptor", "com.lee.config.interceptor.CounterInterceptor");
		properties.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, incerceptors);
		return properties;
	}

	/**
	 * 正常发送
	 *
	 * @param producer 生产商
	 * @param topic    主题
	 * @param message  消息
	 */
	public static void normalSend(KafkaProducer<String, String> producer, String topic, String message) {
		producer.send(new ProducerRecord<>(topic, message));
	}

	/**
	 * 生产者发送与回调
	 *
	 * @param producer 生产商
	 * @param topic    主题
	 * @param message  消息
	 */
	public static void producerSendWithCallback(KafkaProducer<String, String> producer, String topic, String message) {
		producer.send(new ProducerRecord<>(topic, message), (metadata, exception) -> {
			if (Objects.isNull(exception)) {
				System.out.println(metadata.partition() + "--" + metadata.offset());
				System.out.println(metadata.toString());
			} else {
				exception.printStackTrace();
			}
		});
	}

	/**
	 * 生产者发送与分区进行回调
	 *
	 * @param producer 生产商
	 * @param topic    主题
	 * @param message  消息
	 */
	public static void producerSendWithCallbackWithPartition(KafkaProducer<String, String> producer, String topic, String message) {
		producer.send(new ProducerRecord<>(topic, 0, "hello", message), (metadata, exception) -> {
			if (Objects.isNull(exception)) {
				System.out.println(metadata.partition() + "--"
						+ metadata.offset());
				System.out.println(metadata.toString());
			} else {
				exception.printStackTrace();
			}
		});
	}


	/**
	 * 正常发送diy分区
	 *
	 * @param producer 生产商
	 * @param topic    主题
	 * @param message  消息
	 */
	public static void normalSendDiyPartition(KafkaProducer<String, String> producer, String topic, String message) {
		producer.send(new ProducerRecord<>(topic, message));
	}

	/**
	 * 同步发送, 用于保证全局有序.
	 *
	 * @param producer 生产商
	 * @param topic    主题
	 * @param message  消息
	 * @throws ExecutionException   执行异常
	 * @throws InterruptedException 中断异常
	 */
	public static void syncSend(KafkaProducer<String, String> producer, String topic, String message) throws ExecutionException, InterruptedException {
		Future<RecordMetadata> future = producer.send(new ProducerRecord<>(topic, message));
		future.get();
	}
}
